//! High level storage interface that wraps all the other data stores in an
//! opaque interface that does high level operations.  This is intended to be
//! used thorughout most of the stack instead of having handles to individual
//! databases littered all over the place and lets us swap out impls more
//! easily.
//!
//! Many of these are going to be direct wrappers around the underlying
//! datastores, but some of them involve more advanced operations.  The purpose
//! of this is to be an authoritative source on how data management policies
//! are configured and executed, with lower-level details delegated to the
//! underlying datastores and higher-level context management be implemented by
//! the downstream consumers of this type.

use std::sync::Arc;

use aspect_proto::errors::AspProtoError;
use sea_orm::DatabaseConnection;
use thiserror::Error;
use tracing::*;

use aspect_core::message_queue::{MessageQueueId, QueueEntry};
use aspect_db_common::{
    async_mq::AsyncMqDatastore,
    device::{DeviceDatastore, DirEntry},
    device_msg_queue::{DeviceMqDatastore, QueueStatus},
    user::UserDatastore,
};
use aspect_db_llmq::LlmqDatabase;
use aspect_ident::{DeviceIdent, DirPackage};

use crate::constants::*;

#[derive(Debug, Error)]
pub enum StorageError {
    #[error("device {0} already registered")]
    DeviceAlreadyRegistered(DeviceIdent),

    #[error("device {0} does not belong to specified user")]
    UnownedDevice(DeviceIdent),

    #[error("db: {0}")]
    AspectDb(#[from] aspect_db_common::errors::Error),
}

pub struct HomeserverStorage {
    user_ds: Arc<UserDatastore>,
    device_ds: Arc<DeviceDatastore>,
    dmq_ds: Arc<DeviceMqDatastore>,
}

impl HomeserverStorage {
    /// Startup storage tasks.  This just creates some basic databate structures
    /// if they don't already exist, but we might expand this into a more developed
    /// system in the future.
    pub async fn exec_startup_tasks(&self) -> Result<(), StorageError> {
        let mut new_queues = 0;

        let notif_id = MessageQueueId::Server(DMQID_SERVER_NOTIFICATIONS.to_string());
        if self.dmq_ds.create_queue(notif_id.clone()).await? {
            new_queues += 1;
        }

        debug!(%new_queues, "created server queues");

        Ok(())
    }

    /// Creates a user and any other relevant records they might need.
    pub async fn create_user(&self, username: &str, password: &str) -> Result<(), StorageError> {
        self.user_ds.create_user(username, password).await?;
        Ok(())
    }

    /// Checks if a user's password is valid.
    pub async fn check_user_password(
        &self,
        username: &str,
        password: &str,
    ) -> Result<(), StorageError> {
        if let Err(e) = self.user_ds.check_user_password(username, password).await {
            warn!(%username, "invalid password!");
            return Err(e.into());
        }
        Ok(())
    }

    /// Returns if the specified device belongs to the user in question.
    pub async fn check_device_belongs_to_user(
        &self,
        username: &str,
        device: &DeviceIdent,
    ) -> Result<bool, StorageError> {
        // TODO Make this query directly if the entry is there instead of
        // fetching all the devices.
        trace!(%username, %device, "checking device membership"); // TODO remove
        Ok(self
            .user_ds
            .check_user_owns_device(username, device)
            .await?)
    }

    /// Checks if the provided password matches the user and that the device belongs to the user.
    pub async fn check_device_login(
        &self,
        username: &str,
        password: &str,
        ident: &DeviceIdent,
    ) -> Result<(), StorageError> {
        debug!(%username, %ident, "checking ident creds");
        self.check_user_password(username, password).await?;
        if !self.check_device_belongs_to_user(username, ident).await? {
            warn!(%username, %ident, "ident not owned by user!");
            return Err(StorageError::UnownedDevice(*ident));
        }
        Ok(())
    }

    /// Accepts a new DIR package and registers it to a user.
    pub async fn register_device_with_user(
        &self,
        username: &str,
        pkg: DirPackage,
    ) -> Result<(), StorageError> {
        let ident = *pkg.last_update().ident();
        let user_id = self.user_ds.find_user_id_ok(&username).await?;

        // First add the device to the user so that it's locked in.
        self.device_ds.add_device_to_user(&ident, user_id).await?;

        // Then we can insert the actual device entry.
        let de = DirEntry::Pkg(pkg);
        let res = self.device_ds.try_insert_dir_entry(&de).await?;

        if res.is_some() {
            return Err(StorageError::DeviceAlreadyRegistered(ident));
        }

        // And while we're here we can insert the queues the device needs.
        // TODO Maybe do this elsewhere?
        self.setup_device_queues(&ident).await?;

        Ok(())
    }

    /// Creates a high-level message queue if it does not already exist.
    pub async fn create_queue_ifne(&self, mqid: &MessageQueueId) -> Result<(), StorageError> {
        self.dmq_ds.create_queue(mqid.clone()).await?;
        Ok(())
    }

    /// Optimistically sets up the device's queues.  If they're already
    /// subscribed to a queue, does not create another subscription.
    pub async fn setup_device_queues(&self, ident: &DeviceIdent) -> Result<(), StorageError> {
        let mut new_queues = 0;
        let mut new_subs = 0;

        // Create a device message queue if one doesn't exist already.
        let personal_id = MessageQueueId::Device(*ident);
        if self.dmq_ds.create_queue(personal_id.clone()).await? {
            new_queues += 1;
        }

        // Register the device to its inbox queue.
        if self.dmq_ds.sub_device_to_queue(&personal_id, ident).await? {
            new_subs += 1;
        }

        // Register the device the the server notification queue.
        let notif_id = MessageQueueId::Server(DMQID_SERVER_NOTIFICATIONS.to_string());
        if self.dmq_ds.sub_device_to_queue(&notif_id, ident).await? {
            new_subs += 1;
        }

        if new_queues > 0 {
            debug!(%ident, %new_queues, "created new device queues");
        }

        if new_subs > 0 {
            debug!(%ident, %new_subs, "subscribed device to queues");
        }

        Ok(())
    }

    /// Queries a DIR from the database, if it exists.
    pub async fn query_dir_entry(
        &self,
        id: &DeviceIdent,
    ) -> Result<Option<DirEntry>, StorageError> {
        Ok(self.device_ds.query_dir_entry(id).await?)
    }

    /// Tries to accept the dir package, returning an error if unsuccessful or a
    /// better entry if one is found in the database.
    pub async fn try_accept_dir_package(
        &self,
        pkg: DirPackage,
    ) -> Result<Option<DirEntry>, StorageError> {
        let ent = DirEntry::Pkg(pkg);
        let res = self.device_ds.try_insert_dir_entry(&ent).await;

        match res {
            Ok(ent) => Ok(ent),
            Err(e) => {
                warn!(err = %e, "failed to insert DIR entry");
                Err(e.into())
            }
        }
    }

    /// Returns a vec of the device idents linked to the user.
    pub async fn get_user_device_ids(&self, user: &str) -> Result<Vec<DeviceIdent>, StorageError> {
        Ok(self.user_ds.get_user_devices(user).await?)
    }

    /// Dumps all known valid devices.
    pub async fn dump_devices(&self) -> Result<Vec<DirPackage>, StorageError> {
        let ents = self.device_ds.dump_valid_devices(true).await?;
        let mut pkgs = Vec::new();
        for ent in ents {
            if let DirEntry::Pkg(pkg) = ent {
                pkgs.push(pkg);
            }
        }
        Ok(pkgs)
    }

    /// Gets the ID and status of all of the queues a device has access to.
    pub async fn get_device_queues(
        &self,
        ident: &DeviceIdent,
    ) -> Result<Vec<QueueStatus>, StorageError> {
        Ok(self.dmq_ds.get_device_queues(ident).await?)
    }

    /// Queues a message to a device queue.
    pub async fn queue_device_msg(
        &self,
        mqid: &MessageQueueId,
        payload: &[u8],
    ) -> Result<u64, StorageError> {
        let idx = self.dmq_ds.submit_msg_to_queue(mqid, payload).await?;
        Ok(idx)
    }

    /// Fetches messages for a device's queue, checking access permissions.
    pub async fn query_device_queue_msgs(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
        start_idx: u64,
        end_idx: u64,
    ) -> Result<Vec<QueueEntry>, StorageError> {
        let len = if end_idx <= start_idx {
            // TODO probably still add check here
            return Ok(Vec::new());
        } else {
            end_idx - start_idx
        };

        // TODO expose a version of this with a max payload size
        Ok(self
            .dmq_ds
            .query_device_queue_range(device, mqid, start_idx, len, None)
            .await?)
    }

    /// Gets the "next message index" from the queue, which can be set as the
    /// "end" of a query range to fetch all of the messages in the queue.
    pub async fn get_device_queue_next_idx(
        &self,
        device: &DeviceIdent,
        mqid: &MessageQueueId,
    ) -> Result<u64, StorageError> {
        Ok(self.dmq_ds.get_queue_next_idx(device, mqid).await?)
    }
}

/// Opens a storage instance on top of primitives.
pub fn open_on_primitives(
    sled_db: sled::Db,
    dbc: DatabaseConnection,
) -> anyhow::Result<HomeserverStorage> {
    // Set up queue datastores, this involves several layers that we should
    // refactor to make more unified.
    let dmq_tree = sled_db.open_tree("device_mq")?;
    let llmq_ds = LlmqDatabase::new(dmq_tree)?;
    let hlmq_ds = Arc::new(AsyncMqDatastore::new(llmq_ds)?);
    let dmq_ds = Arc::new(DeviceMqDatastore::new(hlmq_ds, dbc.clone())?);

    // The more conventional datastores.
    let user_ds = Arc::new(UserDatastore::new(dbc.clone())?);
    let device_ds = Arc::new(DeviceDatastore::new(dbc.clone())?);

    Ok(HomeserverStorage {
        user_ds,
        device_ds,
        dmq_ds,
    })
}

impl Into<AspProtoError> for StorageError {
    fn into(self) -> AspProtoError {
        // TODO make this reporting more sophisticated
        AspProtoError::Other(self.to_string())
    }
}
